package e3dcs

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"gitee.com/ameise84/e3dcs/internal/discov"
	"gitee.com/ameise84/e3dcs/internal/help"
	"gitee.com/ameise84/e3dcs/internal/log"
	"gitee.com/ameise84/e3dcs/internal/message"
	"gitee.com/ameise84/e3lock"
	"gitee.com/ameise84/e3log"
	"gitee.com/ameise84/e3utils/common"
	"gitee.com/ameise84/e3utils/load_balancer"
	"gitee.com/ameise84/e3utils/uuid"
	"github.com/pkg/errors"
)

func newRouter(serviceName string, mode BLMode, localService service) *router {
	r := &router{
		serviceName:  serviceName,
		mode:         mode,
		vCount:       1,
		bl:           load_balancer.NewBalancer[service](mode),
		stopChan:     make(chan struct{}, 1),
		isLocal:      localService != nil,
		localService: localService,
	}
	if r.isHashLoadBalancer() {
		r.vCount = maxVirtualCount
	}
	return r
}

type router struct {
	common.Service
	serviceName  string
	mode         BLMode
	vCount       uint64
	isLocal      bool
	localService service
	mu           e3lock.SpinLock //用于路由的变更和启动actor
	bl           load_balancer.Balancer[service]
	dcsLock      sync.Locker
	stopChan     chan struct{}
	waitCount    atomic.Int64
}

func (ts *router) E3LogMarshall() string {
	return fmt.Sprintf(`"router":{"service":%s,"islocal":%t}`, ts.serviceName, ts.isLocal)
}

func (ts *router) OnWatch(events []discov.Event) {
	for _, evt := range events {
		f, err := takeServiceWithJson(evt.Value, ts.localService)
		if err != nil {
			log.GetLogger().Error("on watch service with json").Str("js", evt.Value).Println()
			continue
		}

		if f.isLocalService() {
			if evt.Kind == discov.EventDel {
				close(ts.stopChan)
			}
			return
		}

		if !ts.IsRunning() {
			return
		}

		switch evt.Kind {
		case discov.EventCreate:
			ts.onRegisterEvent(f)
		case discov.EventModify:
			ts.onModifyEvent(f)
		case discov.EventDel:
			ts.onUnregisterEvent(f)
		}
	}
}

func (ts *router) post(msg *message.Message) {
	switch msg.GetKind() {
	case message.MsgBroadcastToActor, message.MsgBroadcastToService:
		if ts.localService != nil {
			ts.localService.onBroadcast(msg)
			return
		}
		doMsgPostErr(msg, errors.New("router on broadcast msg not find local service"))
		return
	default:
	}

	if msg.IsToFix() {
		ts.onPostToFix(msg)
	} else {
		ts.onPostToAny(msg)
	}
}

func (ts *router) broadcast(msg *message.Message) bool {
	ts.mu.Lock()
	defer ts.mu.Unlock()

	ts.bl.Range(func(ctx service) bool {
		if ctx.isLocalService() {
			return true
		}
		nMsg := message.TakeMessage()
		nMsg.Clone(msg)
		nMsg.SetToActor(ctx.getServiceID(), ctx.getSessionID(), "")
		if ctx.isLocalService() {
			ctx.onBroadcast(nMsg)
		} else {
			ctx.onPost(nMsg)
		}
		return true
	})
	message.FreeMessage(msg)
	return true
}

func (ts *router) isHashLoadBalancer() bool {
	return ts.mode == BLModeHash
}

func (ts *router) startLocalService(rpcAddr string) error {
	err := ts.Service.Start(func() (err error) {
		ts.dcsLock = discov.GetLocker(ts.serviceName)
		ts.dcsLock.Lock()
		defer func() {
			if err != nil || ts.localService.getState() == stateReady {
				ts.dcsLock.Unlock()
			}
		}()
		serviceName := ts.serviceName
		f := ts.localService
		ts.waitCount.Store(0)

		var sessionID string
		initStat := stateReady
		if ts.isHashLoadBalancer() {
			initStat = statePreparing
		}
		f.(*entity).meta.State = initStat

		f.setRpcAddr(rpcAddr)
		ids := make(map[uint64]struct{}, ts.vCount)
		try := 0
		for {
			ok := false
			clear(ids)
			sessionID = uuid.NewShortV4()
			f.setSessionID(sessionID)
			fs := ts.takeVirtualServices(f)
			for _, vf := range fs {
				ids[vf.LoadBalancerID()] = struct{}{}
			}

			if uint64(len(ids)) != ts.vCount {
				continue
			}

			ok, err = discov.RegisterServiceHashID(serviceName, ids)
			if err != nil {
				return err
			} else if !ok {
				try++
				time.Sleep(time.Duration(try) * 100 * time.Millisecond)
				continue
			}
			break
		}

		ts.registerService(f)
		var (
			rid   int64
			infos []string
			nf    service
		)
		rid, infos, err = discov.GetServices(serviceName) //获取已注册的节点
		if err != nil {
			err = errors.WithMessage(err, "init local service get from svc_disco")
			return err
		}

		for _, v := range infos {
			nf, err = takeServiceWithJson(v, f)
			if err != nil {
				err = errors.Wrap(err, "init local service new from json")
				return err
			}
			if nf.getBLMode() != f.getBLMode() {
				err = errors.New("init local service mode not match")
				return err
			}
			ts.registerService(nf)
		}

		if len(infos) == 0 && initStat == statePreparing {
			ts.updateServiceState(f, stateReady)
		} else {
			ts.waitCount.Add(int64(len(infos)))
		}
		if err = ts.watch(rid); err != nil {
			err = errors.WithMessage(err, "init local service watch")
			return err
		}
		return discov.RegisterService(serviceName, f.getSessionID(), f.metaToJson())
	})
	if err == nil {
		log.GetLogger().ColorMsg(e3log.LevelInfo, e3log.SelectColor(ts.serviceName), "init local service success").Object(ts.localService).Println()
	}
	return err
}

func (ts *router) stopLocalService() {
	ts.Service.Stop(func() {
		//通知暂停
		ts.updateServiceState(ts.localService, stateDestroy)
		ids := make(map[uint64]struct{}, ts.vCount)
		fs := ts.takeVirtualServices(ts.localService)
		for _, vf := range fs {
			ids[vf.LoadBalancerID()] = struct{}{}
		}

		ts.localService.close(ts.bl.Count() > int(ts.vCount))

		ts.unregisterService(ts.localService) //删除路由

		//注销服务
		err := discov.UnregisterService(ts.serviceName, ts.localService.getSessionID())
		if err != nil {
			log.GetLogger().Error("unregister service").Err(err).Println()
			return
		}

		_, _ = <-ts.stopChan //等待注销监听
		err = discov.UnregisterServiceHashID(ts.serviceName, ids)
		if err != nil {
			log.GetLogger().Error("unregister service hash id").Err(err).Println()
		}
	})
}

func (ts *router) registerService(f service) {
	log.GetLogger().Info("register service").Object(f).Str("state", serviceStateString[f.getState()]).Bool("router", ts.isLocal).Bool("local", f.isLocalService()).Println()
	ts.mu.Lock()
	defer ts.mu.Unlock()
	fs := ts.takeVirtualServices(f)
	for _, vf := range fs {
		_ = ts.bl.Register(vf)
	}
}

func (ts *router) unregisterService(f service) {
	if f.isLocalService() {
		log.GetLogger().ColorMsg(e3log.LevelInfo, e3log.SelectColor(ts.serviceName), "unregister local service").Object(f).Str("state", serviceStateString[f.getState()]).Bool("router", ts.isLocal).Bool("local", f.isLocalService()).Println()
	} else {
		log.GetLogger().Info("unregister service").Object(f).Str("state", serviceStateString[f.getState()]).Bool("router", ts.isLocal).Bool("local", f.isLocalService()).Println()
	}
	ts.mu.Lock()
	defer ts.mu.Unlock()
	fs := ts.takeVirtualServices(f)
	for _, vf := range fs {
		_, _ = ts.bl.UnRegister(vf.LoadBalancerID())
	}
}

func (ts *router) updateServiceState(f service, state stateKind) {
	f.setState(state)
	if err := discov.UpdateService(f.getServiceID(), f.getSessionID(), f.metaToJson()); err != nil {
		log.GetLogger().Error("update service state").Object(f).Err(err).Println()
	}
}

func (ts *router) onRegisterEvent(f service) {
	if !ts.isLocal {
		ts.registerService(f)
		return
	}

	toMove := false
	if ts.isHashLoadBalancer() {
		fs := ts.takeVirtualServices(f)
		for _, vf := range fs { //检查本地节点是否需要转移数据
			ctx, err := ts.bl.TakeNext(vf.LoadBalancerID())
			if err == nil && ctx.isLocalService() {
				toMove = true
				ts.updateServiceState(ts.localService, stateMoving)
				break
			}
		}
	}
	ts.registerService(f)
	if ts.isHashLoadBalancer() {
		if toMove {
			ts.localService.moveActor(func(a *actorLocal) bool {
				s, err := ts.findService(a.clerkID)
				return err != nil || !s.isLocalService()
			})
		}
		ts.updateServiceState(ts.localService, stateReady)
	}
}

func (ts *router) onModifyEvent(f service) {
	state := f.getState()
	log.GetLogger().Debug("on modify event").Object(f).Str("state", serviceStateString[state]).Println()
	rf, err := ts.findFixService(f.getSessionID())
	if err != nil {
		if ts.isLocal {
			log.GetLogger().Error("router on modify").Object(f).Err(err).Println()
			return
		}
	} else {
		rf.setState(state)
	}
	switch state {
	case stateMoving:
	case stateReady:
		if ts.isLocal && ts.localService.getState() == statePreparing {
			if ts.waitCount.Add(-1) == 0 {
				ts.dcsLock.Unlock()
				ts.updateServiceState(ts.localService, stateReady)
			}
		}
	case stateDestroy:
	default:
	}
}

func (ts *router) onUnregisterEvent(f service) {
	ts.unregisterService(f) //本地服务已经提前删除路由
}

func (ts *router) watch(rid int64) error {
	return discov.Watch(ts, ts.serviceName, rid)
}

func (ts *router) takeVirtualServices(f service) []service {
	fs := make([]service, ts.vCount)
	fs[0] = f
	for i := uint64(1); i < ts.vCount; i++ {
		v := newServiceVirtual(f, i)
		fs[i] = v
	}
	return fs
}

func (ts *router) findFixService(id string) (service, error) {
	if id == "" {
		panic("hash router find service id is empty")
	}
	hashID := help.Hash16(id)
	ctx, ok := ts.bl.TakeByID(hashID)
	if !ok {
		return nil, errors.New("fix service not found")
	}
	return ctx, nil
}

func (ts *router) findService(id string) (service, error) {
	switch ts.mode {
	case BLModeHash:
		if id == "" {
			panic("hash router find service id is empty")
		}
	default:
	}
	hashID := help.Hash16(id)
	ctx, err := ts.bl.TakeNext(hashID)
	if err != nil {
		return nil, err
	}
	return ctx, nil
}

func (ts *router) onPostToFix(msg *message.Message) {
	ts.mu.Lock()
	defer ts.mu.Unlock()

	s, err := ts.findFixService(msg.GetToActor().SessionID())
	if err != nil {
		doMsgPostErr(msg, err)
		return
	}
	s.onPost(msg)
}

func (ts *router) onPostToAny(msg *message.Message) {
	ts.mu.Lock()
	defer ts.mu.Unlock()

	if ts.isLocal && !ts.isHashLoadBalancer() {
		ts.localService.onPost(msg)
		return
	}
	s, err := ts.findService(msg.GetToActor().ClerkID())
	if err != nil {
		doMsgPostErr(msg, err)
		return
	}
	s.onPost(msg)
}

func (ts *router) wait() {
	ts.mu.Lock()
	defer ts.mu.Unlock()

	//循环所有的负载均衡器,保证所有消息已经顺利投递
	ts.bl.Range(func(s service) bool {
		s.wait()
		return true
	})
}
